package com.atguigu.gmall.realtime.dwd.app;

import com.atguigu.gmall.realtime.common.base.BaseSqlApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.FlinkSqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * 交易域退款陈公公事务事实表
 */
public class DwdTradeRefundPaySucDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeRefundPaySucDetail().start(10018, 4, Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS);
    }

    @Override
    public void handle(StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        // 设置状态的TTL
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));


        // 读取topic_db的数据
        readOdsTopicDb(tableEnv, Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS);

        // 筛选出退款表中退款成功的数据
        Table refundPayment = tableEnv.sqlQuery(
                "select " +
                        " data['id'] id," +
                        " data['order_id'] order_id," +
                        " data['sku_id'] sku_id," +
                        " data['payment_type'] payment_type," +
                        " data['total_amount'] total_amount," +
                        " pt, " +
                        " ts " +
                        " from topic_db " +
                        " where `database`='gmall' " +
                        " and `table`='refund_payment' " +
                        " and `type`='update' " +
                        " and `old`['refund_status'] is not null " +
                        " and `data`['refund_status']='1602'");
        tableEnv.createTemporaryView("refund_payment", refundPayment);

        // 筛选出退单表中退款成功的数据
        Table orderRefundInfo = tableEnv.sqlQuery(
                "select " +
                        " data['order_id'] order_id," +
                        " data['sku_id'] sku_id," +
                        " data['refund_num'] refund_num " +
                        " from topic_db " +
                        " where `database`='gmall' " +
                        " and `table`='order_refund_info' " +
                        " and `type`='update' " +
                        " and `old`['refund_status'] is not null " +
                        " and `data`['refund_status']='0705'");
        tableEnv.createTemporaryView("order_refund_info", orderRefundInfo);


        // 筛选出订单表中退款成功的数据
        Table orderInfo = tableEnv.sqlQuery(
                "select " +
                        " data['id'] id," +
                        " data['user_id'] user_id," +
                        " data['province_id'] province_id " +
                        " from topic_db " +
                        " where `database`='gmall' " +
                        " and `table`='order_info' " +
                        " and `type`='update' " +
                        " and `old`['order_status'] is not null " +
                        " and `data`['order_status']='1006'");
        tableEnv.createTemporaryView("order_info", orderInfo);

        // 读取字典维度表
        // readDimBase


        // join
        Table result = tableEnv.sqlQuery(
                "select " +
                        " rp.id," +
                        " oi.user_id," +
                        " rp.order_id," +
                        " rp.sku_id," +
                        " oi.province_id," +
                        " rp.payment_type," +
                        " dic.info.dic_name payment_type_name," +
                        " ori.refund_num," +
                        " rp.total_amount," +
                        " rp.ts " +
                        " from refund_payment rp " +
                        " join order_refund_info ori " +
                        " on rp.order_id=ori.order_id and rp.sku_id=ori.sku_id " +
                        " join order_info oi " +
                        " on rp.order_id=oi.id " +
                        " join dim_base_dic for system_time as of rp.pt as dic " +
                        " on rp.payment_type=dic.dic_code ");


        // 写出
        tableEnv.executeSql("create table dwd_trade_refund_pay_suc_detail(" +
                " id string," +
                " user_id string," +
                " order_id string," +
                " sku_id string," +
                " province_id string," +
                " payment_type_code string," +
                " payment_type_name string," +
                " date_id string," +
                " callback_time string," +
                " refund_num string," +
                " refund_amount string," +
                " ts bigint , " +
                " PRIMARY KEY(id) NOT ENFORCED" +
                " )" + FlinkSqlUtil.getUpsertKafkaSinkDDl(Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS));

        result.executeInsert(Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS);

    }
}
